package dnet

import (
	"dgo/goutils/utils"
	"encoding/base64"
	"fmt"
	"golang.org/x/net/proxy"
	"io"
	"net"
	"net/url"
	"sync"
	"sync/atomic"
	"time"
)

type TcpMsgCli struct {
	disflag   int32
	stopflag  int32
	disreason string

	closewg  sync.WaitGroup
	execflag int8

	dialProxyAddr string
	lastDialAddr  string
	addr          string
	subtopic      string

	conn net.Conn

	// 最小重连间隔
	reconnectMinInterval time.Duration
	lastConnectTime      time.Time

	// buf 可以进行异步
	OnRecvBuf       func(sender *TcpMsgCli, buf []byte)
	OnBeforeConnect func(sender *TcpMsgCli) bool
	OnConnectFail   func(sender *TcpMsgCli, err error)
	OnConnected     func(sender *TcpMsgCli)
	OnDisconnected  func(sender *TcpMsgCli, reason string)
}

var (
	tcpmsg_cmd_end_bytes = []byte{0x0D, 0x0A, 0x0D, 0x0A}
)

func NewTcpMsgCli() *TcpMsgCli {
	rval := &TcpMsgCli{
		reconnectMinInterval: time.Second * 5,
	}
	return rval
}

func (this *TcpMsgCli) Connected() bool {
	return this.conn != nil && this.disflag == 0
}

func (this *TcpMsgCli) doConnectFail(err error) {
	evt := this.OnConnectFail
	if evt != nil {
		evt(this, err)
	}
}

func (this *TcpMsgCli) doDisconnected() {
	evt := this.OnDisconnected
	if evt != nil {
		evt(this, this.disreason)
	}
}

func (this *TcpMsgCli) doRecvBuf(buf []byte) {
	evt := this.OnRecvBuf
	if evt != nil {
		evt(this, buf)
	}
}

func (this *TcpMsgCli) doConnected() {
	evt := this.OnConnected
	if evt != nil {
		evt(this)
	}
}

func (this *TcpMsgCli) doBeforeConnect() bool {
	if time.Since(this.lastConnectTime) >= this.reconnectMinInterval {
		evt := this.OnBeforeConnect
		if evt != nil {
			return evt(this)
		}
		return true
	} else { // 重连时间过短
		return false
	}
}

/*
配置代理拨号地址
*/
func (this *TcpMsgCli) ConfigDialAddr(s string) bool {
	if this.dialProxyAddr != s {
		this.dialProxyAddr = s
		return true
	}
	return false
}

func (this *TcpMsgCli) ConfigReconnectMinInterval(interval time.Duration) {
	this.reconnectMinInterval = interval
	if this.reconnectMinInterval < time.Millisecond*100 {
		this.reconnectMinInterval = time.Second
	}
}

func (this *TcpMsgCli) ConfigRemoteAddr(s string) bool {
	if this.addr != s {
		this.addr = s
		return true
	}
	return false
}

func (this *TcpMsgCli) ConfigSubTopic(s string) bool {
	if this.subtopic != s {
		this.subtopic = s
		return true
	}
	return false
}

func (this *TcpMsgCli) connect() (err error) {
	this.lastConnectTime = time.Now()
	proxyaddr := this.dialProxyAddr
	if len(proxyaddr) > 0 {
		this.lastDialAddr = fmt.Sprintf("%s,%s", proxyaddr, this.addr)
		d, err := proxy.SOCKS5("tcp", proxyaddr, nil, proxy.Direct)
		if err != nil {
			return err
		}
		conn, err := d.Dial("tcp", this.addr)
		if err != nil {
			return err
		}
		this.conn = conn
	} else {
		this.lastDialAddr = fmt.Sprintf("%s", this.addr)
		conn, err := net.Dial("tcp", this.addr)
		if err != nil {
			return err
		}
		this.conn = conn
	}

	atomic.StoreInt32(&this.disflag, 0)
	return nil
}

func (this *TcpMsgCli) RequestDis(reason string) {
	if atomic.CompareAndSwapInt32(&this.disflag, 0, 1) {
		this.disreason = reason
	}
	conn := this.conn
	if conn != nil {
		conn.Close()
	}
}

func (this *TcpMsgCli) WriteBuf(buf []byte) (n int, err error) {
	conn := this.conn
	if conn == nil {
		return -1, io.ErrClosedPipe
	}
	n, err = conn.Write(buf)
	if err != nil {
		this.RequestDis(err.Error())
	}
	return
}

func PubPack(topic string, data string, pack string) []byte {
	if pack == "base64" {
		data = base64.RawStdEncoding.EncodeToString([]byte(data))
	} else if pack == "url" {
		data = url.QueryEscape(data)
	} else if pack == "json" {
		data = utils.EscapeJsonStr(data)
	}
	strPack := fmt.Sprintf("action:pub\r\ntopic:%s\r\ndata:%s\r\npack:%s\r\n\r\n", topic, data, pack)
	return []byte(strPack)
}

func (this *TcpMsgCli) PubPack(topic string, data string, pack string) []byte {
	return PubPack(topic, data, pack)
}

func (this *TcpMsgCli) Pub(topic string, data string, pack string) (err error) {
	buf := this.PubPack(topic, data, pack)
	_, err = this.WriteBuf(buf)
	return err
}

/*
请求停止
*/
func (this *TcpMsgCli) RequestStop(reason string) {
	this.stopflag = 1
	this.RequestDis(reason)
}

func (this *TcpMsgCli) Stop() {
	this.RequestStop("请求关闭(Stop)")
}

/*
不要在回调事件中回调
*/
func (this *TcpMsgCli) StopAndWait() {
	this.Stop()
	this.closewg.Wait()
}

func (this *TcpMsgCli) SubTopic(topic string) {
	str := fmt.Sprintf("action:sub\r\ntopic:%s\r\n\r\n", topic)
	this.WriteBuf([]byte(str))
}

func (this *TcpMsgCli) Start() {
	this.execflag = 1
	this.disflag = 0
	this.stopflag = 0
	this.closewg.Add(1)
	defer func() {
		this.execflag = 0
		this.closewg.Done()
	}()
	for {
		if this.stopflag == 1 {
			break
		}

		if !this.Connected() {
			if !this.doBeforeConnect() {
				time.Sleep(time.Millisecond * 100)
				continue
			}

			err := this.connect()
			if err != nil {
				this.doConnectFail(err)
				continue
			}
			this.doConnected()
			subTopic := this.subtopic
			if len(subTopic) > 0 {
				this.SubTopic(subTopic)
			}
		}

		buf := make([]byte, 1024) // 方便进行异步处理
		n, err := this.conn.Read(buf)
		if err != nil {
			this.RequestDis(err.Error())
			this.doDisconnected()
			continue
		}
		this.doRecvBuf(buf[:n])
	}
}
